package com.delta.rafg.core.database;

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.math.BigDecimal;
import java.net.URL;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ParameterMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;

import javax.servlet.ServletContext;

import com.delta.rafg.core.config.ConfigSettings;
import com.delta.rafg.core.database.impl.ACacheDataHandler;
import com.delta.rafg.core.database.impl.ADataHandler;
import com.delta.rafg.core.database.impl.GenericDataHandlerImpl;
import com.delta.rafg.core.database.impl.util.DatabaseUtils;
import com.delta.rafg.core.database.impl.JavaDBCacheDataHandlerImpl;
import com.delta.rafg.core.exception.RAFGException;
import com.delta.rafg.core.log.RAFGLogger;

/**
 * @author RAFG Development Team (D9127520)
 *
 */
public class SynchroHandler {

    /**
     * Flag to turn on local debugging.
     */
    private static boolean LOCAL_DBG = false;

    /**
     *
     */
    private List<String> masterDataSourceIdentifiers;

    private List<IDataHandler> masterDataSources;

    private ACacheDataHandler cacheDataSource;

    /**
     * Holds the value of seedFile property.
     */
    private String seedFile = "rafg_ref_seed.sql";
    
    // Added theDBPathSuffix so JavaDB gets generated with the Server Name as suffix
    private String theDBPathSuffix = System.getProperty("AppSrvName")!=null?System.getProperty("AppSrvName"):"";
    
    // Also, cleaned up the cache_database names by using correct variabled
    private String cacheDatabasePath = "cache_database";
    
    private String slashCacheDatabasePath = "/" + cacheDatabasePath;
    
    private String cacheDatabasePathSlash = cacheDatabasePath + "/";
    
    private String cacheDatabaseAbsolutePath = null;

    private ServletContext context;
    
    public void setServletContext(ServletContext servletContext) {
    	context = servletContext;
    }
    
    public static final String CONFIG_SETTINGS_PREFIX = "com.delta.rafg.core.database.synchro";
    
    public void initialize() throws RAFGException {
        initialize(null);
    }
    /**
     * @throws RAFGException
     *
     */
    public void initialize(ConfigSettings configSettings) throws RAFGException {
        if (configSettings == null) {
        	configSettings = ConfigSettings.getSingleton();
        }
    	initializeMasterDataHandlers(configSettings);
        initializeCacheDataHandler(configSettings);
    }
    
    public void destroy() {
    	if (masterDataSources != null) {
	    	for(IDataHandler h : masterDataSources) {
	    		h.destroy();
	    		h = null;
	    	}
	    	masterDataSources = null;
    	}
    	if (cacheDataSource != null) {
    		cacheDataSource.destroy();
	    	cacheDataSource = null;
    	}
    }

    public void initializeMasterDataHandlers(ConfigSettings configSettings) throws RAFGException {
//        Properties configProps = ConfigSettings.getSingleton();
        String masterDataSourcesProp = configSettings.getProperty(CONFIG_SETTINGS_PREFIX + ".master");
        String[] dsArray = masterDataSourcesProp.split(",");
        masterDataSourceIdentifiers = new ArrayList<String>();
        for (String ds : dsArray) {
        	if(ds.trim().length()>0) {
        		masterDataSourceIdentifiers.add(ds.trim());
        	}
        }
        
        boolean ignoreAllJndi = Boolean.parseBoolean(configSettings.getProperty(ADataHandler.PROP_DATASOURCE_CONFIG_PREFIX + ".ignoreJndi", "false"));

        masterDataSources = new CopyOnWriteArrayList<IDataHandler>();
        for (String id : masterDataSourceIdentifiers) {
        	if (ignoreAllJndi) {
        		configSettings.setProperty(ADataHandler.PROP_DATASOURCE_CONFIG_PREFIX + "." + id + ".ignoreJndi", "true");
        	}
            IDataHandler handler = DataHandlerFactory.getNewDataHandlerImpl(id);
            handler.initialize(id, configSettings);
            masterDataSources.add(handler);
        }
    }

    private void initializeCacheDataHandler(ConfigSettings configSettings) throws RAFGException {
    	ConfigSettings configProps = ConfigSettings.getSingleton();
        String cacheDataSourcesProp = configProps.getProperty(CONFIG_SETTINGS_PREFIX + ".cache");
        cacheDataSource = DataHandlerFactory.getNewCacheDataHandler();
        DataHandlerOptions myOpts = new DataHandlerOptions();

        String seedPath = getCacheDatabaseAbsolutePath();

    	if (seedPath != null) {
    		// Suffix theDBPathSuffix to JavaDB
			String dbPath = seedPath + "/JavaDB" + theDBPathSuffix;
        	myOpts.getDataHandlerSpecificOptionsMap().put("DataBasePath", dbPath);
    	}
    	myOpts.setDataHandlerPropertyIdentifier(cacheDataSourcesProp);
    	myOpts.setConfigSettings(configSettings);
    	cacheDataSource.initialize(cacheDataSourcesProp, myOpts);

    	if (!isCacheCreated()) {
        	loadCacheFromSeed();
        }
    }

    private synchronized String getCacheDatabaseAbsolutePath() throws RAFGException {
    	if (cacheDatabaseAbsolutePath == null) {
        	try {
	        	URL seedPathUrl = this.getClass().getClassLoader().getResource(slashCacheDatabasePath + "/" + seedFile);
	        	RAFGLogger.error("SynchroHandler: slashCacheDatabasePath + seedFile=" + slashCacheDatabasePath + "/" + seedFile);
	
	        	boolean isInJar = false;
	        	if (seedPathUrl != null && seedPathUrl.toString().contains("rafgcore.jar!/")) {
	        		isInJar = true;  //detect if this is inside the rafgcore jar.  This can happen if this is run "on server" in a websphere install inside eclipse.. 
	        	}
	        	
	            if (seedPathUrl != null && !isInJar) {
					cacheDatabaseAbsolutePath = new File(seedPathUrl.getFile()).getCanonicalFile().getParent();
					RAFGLogger.error("SynchroHandler: this.getClass().getClassLoader().getResource(...) result=" + cacheDatabaseAbsolutePath);
	            } else if (context != null) {
	        		//use Rafg.html to locate path to war file root.        	
	            	URL warRoot = context.getResource("/Rafg.html");
					if (warRoot != null) {
						cacheDatabaseAbsolutePath = new File(warRoot.getFile()).getCanonicalFile().getParentFile().getParent() + slashCacheDatabasePath;
					}
					RAFGLogger.error("SynchroHandler: cache path determined from ServletContext=" + cacheDatabaseAbsolutePath);
	        	} else {
		        	seedPathUrl = this.getClass().getClassLoader().getResource(cacheDatabasePathSlash + seedFile);
		        	RAFGLogger.error("SynchroHandler: trying cacheDatabasePathSlash + seedFile=" + cacheDatabasePathSlash + seedFile);
		            if (seedPathUrl != null) {
						cacheDatabaseAbsolutePath = new File(seedPathUrl.getFile()).getCanonicalFile().getParent();
		            }
	        	}
			} catch (IOException e) {
				throw new RAFGException(e);
			}
    	}
    	return cacheDatabaseAbsolutePath;
    }
    
    IDataHandler getCacheDataHandler() {
    	return cacheDataSource;
    }
    
    public List<IDataHandler> getMasterDataHandlers() {
    	return masterDataSources;
    }

//    public void syncAllTables(IDataHandler masterDataHandler, IDataHandler cacheDataHandler) throws SQLException, RAFGException {
    public void syncAllTables() throws SQLException, RAFGException, ClassNotFoundException {
        for (IDataHandler masterDataHandler : masterDataSources) {
            for (String table : masterDataHandler.getMonitoredTables()) {
                syncTable(masterDataHandler.getConnection(), table);
            }
        }
    }

    public Map<IDataHandler, String[]> getChangedMasterTables() throws RAFGException {
        Map<IDataHandler, String[]> changedMasterTables = new Hashtable<IDataHandler, String[]>();
        for (IDataHandler masterDataHandler : masterDataSources) {
        	getChangedMasterTables(changedMasterTables, masterDataHandler);
        }
        return changedMasterTables;
    }

    public Map<IDataHandler, String[]> getChangedMasterTables(IDataHandler masterDataHandler) throws RAFGException {
        Map<IDataHandler, String[]> changedMasterTables = new Hashtable<IDataHandler, String[]>();
        getChangedMasterTables(changedMasterTables, masterDataHandler);
        return changedMasterTables;
    }

    public void getChangedMasterTables(Map<IDataHandler, String[]> changedMasterTables, IDataHandler masterDataHandler) throws RAFGException {
        if (changedMasterTables == null || masterDataHandler == null) {
        	return;
        }
        String[] changedTables = masterDataHandler.getChangedTables();
        if (changedTables != null) {
        	changedMasterTables.put(masterDataHandler, changedTables);
        }
    }

    public void addGenericDataSource(IGenericDataSource dataSource) throws RAFGException {
    	final IDataHandler handler = new GenericDataHandlerImpl(dataSource);
    	DataHandlerOptions opts = new DataHandlerOptions();
    	opts.setDataHandlerPropertyIdentifier(dataSource.getPropertyIdentifier());
    	opts.getDataHandlerSpecificOptionsMap().put(GenericDataHandlerImpl.CACHE_HANDLER_KEY, getCacheDataHandler());
    	handler.initialize(opts);
    	masterDataSources.add(handler);
    	
    	if (dataSource.isSynchronizable()) {
	    	//Uncommented so the generic datasource will get synchronized as soon as it is added. - Tom Prado 
	    	Thread worker = new Thread(new Runnable(){
				@Override
				public void run() {
			    	for(String table : handler.getMonitoredTables()) {
			    		try {
							syncTable(handler.getConnection(), table);
						} catch (SQLException e) {
							//do nothing on this initial sync if it fails.
						} catch (RAFGException e) {
							//do nothing on this initial sync if it fails.
						} catch (ClassNotFoundException e) {
							//do nothing on this initial sync if it fails.
						} catch (Exception e) {
							//do nothing on this initial sync if it fails.
						}
			    	}
				}}); 
	    	worker.setDaemon(true);
	    	worker.start();
    	}
    }
    
    
    /**
     * @throws RAFGException
     */
    private void loadCacheFromSeed() throws RAFGException {
        IDataHandler seedHandler = DataHandlerFactory.getNewSeedDataHandler();
        DataHandlerOptions seedOpts = new DataHandlerOptions();
        //use the same datahandler identifier as the cache table since that is where this is going to get written to.
        seedOpts.setDataHandlerPropertyIdentifier(cacheDataSource.getDataHandlerOptions().getDataHandlerPropertyIdentifier());
        seedOpts.setConfigSettings(ConfigSettings.getSingleton());
        Map<String, Object> seedOptsMap = seedOpts.getDataHandlerSpecificOptionsMap();

//        	URL seedPathUrl = com.delta.rafg.core.database.SynchroHandler.class.getResource(cacheDatabasePath + "/" + seedFile);

       	String cacheSeedPath = getCacheDatabaseAbsolutePath();
        	//determine if I am a .class file or inside a .jar.  If a .class file, I will need to go up
        	//several paths to get to the database_cache directory.

//       	RAFGLogger.info("loadCacheFromSeed() cacheSeedPath=" + cacheSeedPath);
//        	String seedPathAdj;
       	if (cacheSeedPath == null) {
       		throw new RAFGException("Failed to find seed file " + seedFile);
//        	} else if (seedPathUrl.toString().endsWith("rafgcore/com/delta/rafg/core/database/")) {
        		//this is if it's in an expanded .class file running in a hosted websphere in eclipse.  It will be in the above path.
//        		seedPathAdj = "../../../../../cache_database/";
//        	} else if (seedPathUrl.toString().endsWith("src/com/delta/rafg/core/database/")) {
           		//this is if it's in an expanded .class file running in a hosted jetty.  It will be in the above path.
//        		seedPathAdj = "../../../../../../cache_database/";
//        	} else {
        		//this is if it's in a jar in the lib path.
//        		seedPathAdj = "../cache_database/";
//        	}
       	}
        	
//        	String cacheSeedPath = new File(seedPathUrl.getFile()).getCanonicalFile().getParent();
        		
//        	String cacheSeedPath = new File(seedPathUrl.getFile(), seedPathAdj).getCanonicalPath();

       	RAFGLogger.info("loadCacheFromSeed() calculated cacheSeedPath=" + cacheSeedPath);

    	seedOptsMap.put("SeedPath", cacheSeedPath);
    	seedOptsMap.put("SeedFile", seedFile);
    	String serviceName = cacheDataSource.getDataHandlerOptions().getServiceName();
		if (serviceName != null) {
			seedOptsMap.put("ServiceName", cacheDataSource.getDataHandlerOptions().getServiceName());
		}
    	seedOptsMap.put("CacheSqlConnection", cacheDataSource.getConnection());

        seedHandler.initialize(seedOpts);
    }

    /**
     * @return true if the cache database tables have been created already.
     * @throws RAFGException A RAFGException
     */
    private boolean isCacheCreated() throws RAFGException {
    	boolean cacheCreated = true;
    	Statement stmt = null;
    	try {
    		String schemaSuffix = ((ACacheDataHandler) cacheDataSource).getCacheSchemaSuffix();

			Connection conn = cacheDataSource.getConnection();
			stmt = conn.createStatement();

	        for (IDataHandler masterDataHandler : masterDataSources) {
	            for (String srcFullTable : masterDataHandler.getMonitoredTables()) {
	            	String srcSchema = DatabaseUtils.getSchemaName(srcFullTable);
	            	String srcTable = DatabaseUtils.getTableName(srcFullTable);
	            	String cacheTable = srcSchema + schemaSuffix + "." + srcTable;
//	            	RAFGLogger.info("Inspecting table " + cacheTable);
            		if (!cacheDataSource.checkTableExists(stmt, cacheTable)) {
            			cacheCreated = false;
            			RAFGLogger.info(cacheTable + " missing. Recreating reference cache.");
            			break;
            		}
	            }
	            if (!cacheCreated) {
	            	break;
	            }
			}
		} catch (SQLException e) {
			throw new RAFGException(e);
		} finally {
			if (stmt != null) {
				try {
					stmt.close();
				} catch (SQLException e) {
				}
			}
		}

		return cacheCreated;
    }


    /**
     * Syncronize contents of table from sourceConnection to destConnection.
     * sourceConnection is the master and destConnection will be modified to
     * match sourceConnection.
     * @param sourceConnection
     * @param destConnection
     * @param tableName
     * @throws SQLException
     * @throws RAFGException
     * @throws ClassNotFoundException 
     */
    public synchronized void syncTable(final Connection sourceConnection, String fullyQualifiedTableName) throws SQLException, RAFGException, ClassNotFoundException {

        Connection destConnection = null;
        Statement sourceStmt = null;
        Statement destStmt = null;
        ResultSet sourceRs = null;
        ResultSet destRs = null;
        PreparedStatement insertPreparedStatement = null;
        PreparedStatement deletePreparedStatement = null;
        Statement destStmt1 = null;

        try {
        	destConnection = ((JavaDBCacheDataHandlerImpl)cacheDataSource).openConnection();
	        
	        if (sourceConnection instanceof com.delta.rafg.core.database.impl.AGenericDataSource.GenericSqlConnection) {
	//        	LOCAL_DBG = true;
	        }
	
	        DatabaseMetaData sourceMd = sourceConnection.getMetaData();
	        DatabaseMetaData destMd = destConnection.getMetaData();
	
	        if (LOCAL_DBG) {
	        	RAFGLogger.info("syncTable(): Will migrate " + fullyQualifiedTableName
	                    + " from " + sourceMd.getDatabaseProductName()
	                    + " to cached " + destMd.getDatabaseProductName());
	        }
	
	        String destSchemaSuffix = ((ACacheDataHandler) cacheDataSource).getCacheSchemaSuffix();
	        String destSchemaName = DatabaseUtils.getSchemaName(fullyQualifiedTableName);
	        String destTableName = DatabaseUtils.getTableName(fullyQualifiedTableName);
	        String destFullyQualifiedTableName = destSchemaName + destSchemaSuffix + "." + destTableName;
	
	        String sourceProductName = sourceMd.getDatabaseProductName();
	        String destProductName = destMd.getDatabaseProductName();
	
	        List<String> primaryKeys = new LinkedList<String>();
	        List<String> dataColumns = new LinkedList<String>();
	        Map<String, String> columnTypes = new HashMap<String, String>();
	
	        calculateKeysAndDataColumns(primaryKeys, dataColumns, columnTypes, fullyQualifiedTableName, sourceMd);
	
	        List<String> allColumns = new LinkedList<String>();
	        allColumns.addAll(primaryKeys);
	        allColumns.addAll(dataColumns);
	
	        String selectCmdSrc = generateSelectCmd(fullyQualifiedTableName, primaryKeys, dataColumns, columnTypes, sourceProductName);
	        String selectCmdDest = generateSelectCmd(destFullyQualifiedTableName, primaryKeys, dataColumns, columnTypes, destProductName);
	
	/*
	        if (fullyQualifiedTableName.equalsIgnoreCase("CMS.CMS_DBUTIL_HIST")) {
	            LOCAL_DBG = true;
	        } else if (LOCAL_DBG) {
	                LOCAL_DBG = false;
	        }
	*/
	
	        if (LOCAL_DBG) {
	        	RAFGLogger.info("Using SQL to Search " + fullyQualifiedTableName + " source table: " + selectCmdSrc);
	        	RAFGLogger.info("Using SQL to Search " + destFullyQualifiedTableName + " cache table: " + selectCmdDest);
	        }
	
	        int rowsInserted = 0;
	        int rowsUpdated = 0;
	        int rowsDeleted = 0;
	        long rowsMatched = 0;
	
	        sourceStmt = sourceConnection.createStatement();
	        destStmt = destConnection.createStatement();
	        destConnection.setAutoCommit(false);
	        
	        /*
	         * Added logic to remove the CACHE for MAGELLAN.PARTNER.
	         * SYNC logic is failing for this table becoz of no PK defined. 
	         * This code should be removed once PK is defiend for this table.
	         */
	        if(dataColumns.isEmpty() && fullyQualifiedTableName.equalsIgnoreCase("MAGELLAN.PARTNER")){	
	        	RAFGLogger.info("Indide PARTNER Table Sync");
	        	destStmt1 = destConnection.createStatement();
	        	destStmt1.execute("DELETE FROM MAGELLAN_CACHE.PARTNER");	        	
        	}

	
	//        sourceStmt.execute("ALTER SESSION SET NLS_SORT=");
	//        destStmt.execute("ALTER SESSION SET NLS_SORT=ENGLISH");
	
	//        Statement destModifierStmt = destConnection.createStatement();
	
	        sourceRs = sourceStmt.executeQuery(selectCmdSrc);
	        destRs = destStmt.executeQuery(selectCmdDest);
	
	        insertPreparedStatement = destConnection.prepareStatement(
	                generateInsertPreparedStatementSql(destFullyQualifiedTableName, allColumns)  //,
	                );
	        deletePreparedStatement = destConnection.prepareStatement(
	                generateDeletePreparedStatementSql(destFullyQualifiedTableName, primaryKeys)
	                );
	
	        boolean bMoreDestRecs = destRs.next();
	        boolean bMoreSourceRecs = sourceRs.next();
	        while (bMoreSourceRecs || bMoreDestRecs) {
	            RowCompareResult result;
	            if (bMoreSourceRecs && bMoreDestRecs) {
	                result = compareResultSets(sourceRs, destRs, primaryKeys);
	            } else if (bMoreSourceRecs) {
	                result = new RowCompareResult(RowCompareResult.INSERT_SOURCE_ROW);
	            } else {
	                result = new RowCompareResult(RowCompareResult.DELETE_DESTINATION_ROW);
	            }
	
	            if (result.getResult() == RowCompareResult.ROWS_MATCH) {
	/*            	
	                if (LOCAL_DBG) {
	                    RAFGLogger.info("Rows Match");
	                }
	*/                
	                rowsMatched++;
	                bMoreSourceRecs = sourceRs.next();
	                bMoreDestRecs = destRs.next();
	            } else if (result.getResult() == RowCompareResult.INSERT_SOURCE_ROW) {
	                if (LOCAL_DBG) {
	                    RAFGLogger.info("\nComparing rows:    source(" + (bMoreSourceRecs ? DatabaseUtils.dumpRow(sourceRs) : "No Row")
	                            + ") \n                   dest  (" + (bMoreDestRecs ? DatabaseUtils.dumpRow(destRs) : "No Row") + ")");
	                    RAFGLogger.info("Inserting row:     " + DatabaseUtils.dumpRow(sourceRs));
	                }
	                setInsertStatementColumnData(insertPreparedStatement, sourceRs, allColumns);
	//                String sql = generateInsertSql(sourceMd, destMd, sourceRs, destRs, destFullyQualifiedTableName, primaryKeys, dataColumns);
	//                if (LOCAL_DBG) {
	//                    RAFGLogger.info("== SQL ==  " + sql);
	//                }
	                try {
	                    //int cnt = destModifierStmt.executeUpdate(sql);
	                    int cnt = insertPreparedStatement.executeUpdate();
	                    rowsInserted += cnt;
	                    if (LOCAL_DBG) {
	                        RAFGLogger.info("-INSERT SUCCESSFULL- " + cnt + " records inserted.");
	                    }
	                } catch (SQLException e) {
	                	RAFGLogger.error(e, "-INSERT FAILED- Caught: " + e.getMessage() + " :: sourceRs=" + DatabaseUtils.dumpRow(sourceRs) + " :: destRs=" + DatabaseUtils.dumpRow(destRs));
	                }
	                bMoreSourceRecs = sourceRs.next();
	            } else if (result.getResult() == RowCompareResult.DELETE_DESTINATION_ROW) {
	                if (LOCAL_DBG) {
	                    RAFGLogger.info("\nComparing rows:    source(" + (bMoreSourceRecs ? DatabaseUtils.dumpRow(sourceRs) : "No Row")
	                            + ") \n                   dest  (" + (bMoreDestRecs ? DatabaseUtils.dumpRow(destRs) : "No Row") + ")");
	                    RAFGLogger.info("Deleting row:      " + DatabaseUtils.dumpRow(destRs));
	                }
	                setDeleteStatementColumnData(deletePreparedStatement, destRs, primaryKeys);
	//                String sql = generateDeleteSql(destMd, destRs, destFullyQualifiedTableName, primaryKeys);
	//                if (LOCAL_DBG) {
	//                    RAFGLogger.info("== SQL ==  " + sql);
	//                }
	                try {
	                    int cnt = deletePreparedStatement.executeUpdate();
	                    rowsDeleted += cnt;
	                    if (LOCAL_DBG) {
	                        RAFGLogger.info("-DELETE SUCCESSFULL- " + cnt + " records deleted.");
	                    }
	                } catch (SQLException e) {
	                    RAFGLogger.error(e, "-DELETE FAILED- Caught: " + e.getMessage() + " :: sourceRs=" + DatabaseUtils.dumpRow(sourceRs) + " :: destRs=" + DatabaseUtils.dumpRow(destRs));
	                }
	                bMoreDestRecs = destRs.next();
	            } else if (result.getResult() == RowCompareResult.UPDATE_DESTINATION_DATA) {
	                if (LOCAL_DBG) {
	                    RAFGLogger.info("\nComparing rows:    source(" + (bMoreSourceRecs ? DatabaseUtils.dumpRow(sourceRs) : "No Row")
	                            + ") \n                   dest  (" + (bMoreDestRecs ? DatabaseUtils.dumpRow(destRs) : "No Row") + ")");
	                    RAFGLogger.info("Updating row:      " + DatabaseUtils.dumpRow(destRs) + " for columns=" + result.getChangedDataColumns());
	                }
	
	                PreparedStatement updatePreparedStatement = destConnection.prepareStatement(
	                        generateUpdatePreparedStatementSql(destFullyQualifiedTableName, primaryKeys, result)
	                        );
	                setUpdateStatementColumnData(updatePreparedStatement, sourceRs, destRs, result, primaryKeys);
	//                String sql = generateUpdateSql(sourceMd, destMd, sourceRs, destRs, result, destFullyQualifiedTableName, primaryKeys);
	//                if (LOCAL_DBG) {
	//                    RAFGLogger.info("== SQL ==  " + sql);
	//                }
	                try {
	//                    int cnt = destModifierStmt.executeUpdate(sql);
	                    int cnt = updatePreparedStatement.executeUpdate();
	                    rowsUpdated += cnt;
	                    updatePreparedStatement.close();
	                    if (LOCAL_DBG) {
	                        RAFGLogger.info("-UPDATE SUCCESSFULL- " + cnt + " records updated.");
	                    }
	                } catch (SQLException e) {
	                	RAFGLogger.error(e, "-UPDATE FAILED- Caught: " + e.getMessage() + " :: sourceRs=" + DatabaseUtils.dumpRow(sourceRs) + " :: destRs=" + DatabaseUtils.dumpRow(destRs));
	                }
	                bMoreSourceRecs = sourceRs.next();
	                bMoreDestRecs = destRs.next();
	            } else {
	                throw new RAFGException("Unknown result from compareResultSets() call: " + result);
	            }
	        }
	
	        destConnection.commit();
	
	        int len = 40 - fullyQualifiedTableName.length();
	        char[] padding = null;
	        if (len > 0) {
	            padding = new char[len];
	            Arrays.fill(padding, ' ');
	        } else {
	        	padding = new char[0];
	            len = 0;
	        }
	        if (rowsUpdated != 0 || rowsDeleted != 0 || rowsInserted != 0) {
	        	RAFGLogger.info("Done Syncing " + fullyQualifiedTableName + String.valueOf(padding) + " Rows Inserted=" + rowsInserted
		                + "; Updated=" + rowsUpdated + "; Deleted=" + rowsDeleted + "; Unchanged=" + rowsMatched, (Object[]) null);
	
	            cacheDataSource.fireTableChangedEvent(new DataEvent(destFullyQualifiedTableName));
	        }
	
	        if (sourceConnection instanceof com.delta.rafg.core.database.impl.AGenericDataSource.GenericSqlConnection) {
	        	LOCAL_DBG = false;
	        }

//        if (LOCAL_DBG) {
//            RAFGLogger.info("syncTable(): Done Syncing.");
//        }
    	} finally {
    		if (destRs != null) {
    			destRs.close();
    		}
	        if (sourceRs != null) {
	        	sourceRs.close();
	        }
	        if (destStmt != null) {
	        	destStmt.close();
	        }
	        if (destStmt1 != null) {
	        	destStmt1.close();
	        }
	        if (sourceStmt != null) {
		        sourceStmt.close();
	        }
	        if (insertPreparedStatement != null) {
	        	insertPreparedStatement.close();
	        }
	        if (deletePreparedStatement != null) {
	        	deletePreparedStatement.close();
	        }
    		if (destConnection != null) {
    			destConnection.close();
    		}
    	}
        return;
    }


    private String generateSelectCmd(String fullyQualifiedTableName, List<String> primaryKeys,
            List<String> dataColumns, Map<String, String> columnTypes, String productName) {

        StringBuilder selectCmd = new StringBuilder("SELECT ");

        boolean bFirst = true;
        for (String s : primaryKeys) {
            if (!bFirst) {
                selectCmd.append(", ");
            }
            selectCmd.append(s);
            bFirst = false;
        }
        bFirst = true;
        if (dataColumns.size() > 0) {
            selectCmd.append(", ");
        }
        for (String s : dataColumns) {
            if (!bFirst) {
                selectCmd.append(", ");
            }
            selectCmd.append(s);
            bFirst = false;
        }
        selectCmd.append(" FROM " + fullyQualifiedTableName);
        selectCmd.append(" ORDER BY ");
        bFirst = true;
        for (String s : primaryKeys) {
            if (!bFirst) {
                selectCmd.append(", ");
            }

            // if this is a date and it's Oracle,
            //use TO_CHAR to process it because oracle dates
            //are not always "equal" when they should be.
            boolean bUseToCharForDate = false;
            if (columnTypes.get(s).equalsIgnoreCase("DATE")) {
                if (productName.contains("Oracle")) {
                    bUseToCharForDate = true;
                }
            }

            if (bUseToCharForDate) {
                selectCmd.append("TO_CHAR(");
            }
            selectCmd.append(s);
            if (bUseToCharForDate) {
                selectCmd.append(",'YYYYMMDD')");
            }

            bFirst = false;
        }

        return selectCmd.toString();
    }

    /**
     * @param primaryKeys
     * @param dataColumns
     * @param tableName
     * @param sourceMd
     * @throws SQLException
     */
    private void calculateKeysAndDataColumns(List<String> primaryKeys, List<String> dataColumns, Map<String,String> columnTypes, String fullyQualifiedTableName, DatabaseMetaData sourceMd) throws SQLException {

        boolean bNoPrimaryKeysDefined = false;

        //get which columns are part of the primary key
        ResultSet mdPkRs = sourceMd.getPrimaryKeys(null, DatabaseUtils.getSchemaName(fullyQualifiedTableName), DatabaseUtils.getTableName(fullyQualifiedTableName).toUpperCase());

        while (mdPkRs.next()) {
            String name = mdPkRs.getString("COLUMN_NAME").toUpperCase();
            primaryKeys.add(name);
        }
        mdPkRs.close();
        mdPkRs = null;

        //if there are no primary keys, treat all columns as primary keys...
        if (primaryKeys.size() == 0) {
            bNoPrimaryKeysDefined = true;
        }

        //get columns from source
        String schemaName = DatabaseUtils.getSchemaName(fullyQualifiedTableName);
        String tableName = DatabaseUtils.getTableName(fullyQualifiedTableName);
        ResultSet sourceMdRs = sourceMd.getColumns(null, schemaName, tableName, null);
        while (sourceMdRs.next()) {
            String name = sourceMdRs.getString("COLUMN_NAME").toUpperCase();
            int size = sourceMdRs.getInt("COLUMN_SIZE");
            int dataType = sourceMdRs.getInt("DATA_TYPE");
            String dataTypeStr = sourceMdRs.getString("TYPE_NAME").toUpperCase();

            if (LOCAL_DBG) {
            	RAFGLogger.info("Source Column: Name=[" + name + "] Size=["
                        + size + "] type=[" + dataType + "(" + DatabaseUtils.getJdbcTypeName(dataType) + ")] TypeName=[" + dataTypeStr + "] ");
            }

            if (bNoPrimaryKeysDefined
                    && dataType != java.sql.Types.BINARY
                    && dataType != java.sql.Types.CLOB
                    && dataType != java.sql.Types.NCLOB
                    && dataType != java.sql.Types.BLOB
                    && dataType != java.sql.Types.LONGNVARCHAR
                    && dataType != java.sql.Types.LONGVARBINARY
                    && dataType != java.sql.Types.LONGVARCHAR) {
                primaryKeys.add(name);
            } else if (!primaryKeys.contains(name)) {
                //add as a regular data column
                dataColumns.add(name);
            }

            columnTypes.put(name, dataTypeStr);

        }
        sourceMdRs.close();
    }

    /**
     * @param destMd
     * @param destRs
     * @param tableName
     * @param primaryKeys
     * @return SQL Command
     * @throws SQLException
     */
/*
    private String generateDeleteSql(DatabaseMetaData destMd, ResultSet destRs, String tableName, List<String> primaryKeys) throws SQLException {
        final StringBuilder sqlStr = new StringBuilder();
        sqlStr.append("delete from ");
        sqlStr.append(tableName);
        sqlStr.append(" where ");
        boolean bFirst = true;
        for (String s : primaryKeys) {
            if (!bFirst) {
                sqlStr.append(" and ");
            }
            bFirst = false;
            sqlStr.append(s);
            sqlStr.append(translateColumnData(destRs.getObject(s), destMd.getDatabaseProductName(), true));
        }
        return sqlStr.toString();
    }
*/

    /**
     * @param sourceMd
     * @param destMd
     * @param sourceRs
     * @param destRs
     * @param tableName
     * @param primaryKeys
     * @param dataColumns
     * @return SQL Command
     * @throws SQLException
     */
/*
    private String generateInsertSql(DatabaseMetaData sourceMd,
            DatabaseMetaData destMd,
            ResultSet sourceRs,
            ResultSet destRs,
            String tableName,
            Set<String> primaryKeys, Set<String> dataColumns, PreparedStatement preparedStatement) throws SQLException {
        final StringBuilder sqlStr = new StringBuilder();
        HashSet<String> cols = new HashSet<String>(primaryKeys);
        cols.addAll(dataColumns);

        sqlStr.append("insert into ");
        sqlStr.append(tableName);
        sqlStr.append("(");
        boolean bFirst = true;
        for (String s : cols) {
            if (!bFirst) {
                sqlStr.append(",");
            }
            bFirst = false;
            sqlStr.append(s);
        }

        sqlStr.append(") values (");
        bFirst = true;
        for (String s : cols) {
            if (!bFirst) {
                sqlStr.append(",");
            }
            bFirst = false;
            sqlStr.append(translateColumnData(sourceRs.getObject(s), destMd.getDatabaseProductName(), false));
        }
        sqlStr.append(")");

        return sqlStr.toString();
    }
*/

    private void setInsertStatementColumnData(
            PreparedStatement insertPreparedStmt,
            ResultSet sourceRs,
            List<String> allColumns) throws RAFGException {
        int allColsLen = allColumns.size();
        try {
            for (int i = 1; i <= allColsLen; i++) {
                setPreparedStmtColumnData(insertPreparedStmt, i, sourceRs.getObject(i));
            }
        } catch (SQLException e) {
            throw new RAFGException(e);
        }

    }

    private void setDeleteStatementColumnData(
            PreparedStatement deletePreparedStmt,
            ResultSet destRs,
            List<String> primaryKeys) throws RAFGException {
        int allColsLen = primaryKeys.size();
        try {
            for (int i = 1; i <= allColsLen; i++) {
                setPreparedStmtColumnData(deletePreparedStmt, i, destRs.getObject(i));
            }
        } catch (SQLException e) {
            throw new RAFGException(e);
        }
    }

    private void setUpdateStatementColumnData(
            PreparedStatement updatePreparedStmt,
            ResultSet sourceRs,
            ResultSet destRs,
            RowCompareResult result,
            List<String> primaryKeys) throws RAFGException  {
        List<String> changedCols = result.getChangedDataColumns();
        int changedColsLen = changedCols.size();
        int primaryKeysLen = primaryKeys.size();
        try {
            for (int i = 1; i <= changedColsLen; i++) {
                String changedCol = changedCols.get(i-1);
                setPreparedStmtColumnData(updatePreparedStmt, i, sourceRs.getObject(changedCol));
            }
            for (int i = 1; i <= primaryKeysLen; i++) {
                String key = primaryKeys.get(i-1);
                setPreparedStmtColumnData(updatePreparedStmt, changedColsLen + i, destRs.getObject(key));
            }
        } catch (SQLException e) {
            throw new RAFGException(e);
        }
    }

    private void setPreparedStmtColumnData(
            PreparedStatement preparedStmt,
            int column,
            Object data) throws SQLException, RAFGException {
        if (data instanceof oracle.sql.BLOB) {
            java.sql.Blob sourceBlob = (java.sql.Blob) data;
            long blobLen = sourceBlob.length();
            if (blobLen > Integer.MAX_VALUE) {
                throw new RAFGException("Blobs cannot be greater than " + Integer.MAX_VALUE + " in this implementation.");
            }
            byte[] b = new byte[(int) blobLen];
            b = sourceBlob.getBytes(1, (int) blobLen);
            InputStream is = new ByteArrayInputStream(b);
            preparedStmt.setBinaryStream(column, is, (int) blobLen);
        } else {
        	if(data != null) {
        		preparedStmt.setObject(column, data);
        	} else {
        		ParameterMetaData pmd = preparedStmt.getParameterMetaData();
        		int colType = pmd.getParameterType(column);
           		preparedStmt.setNull(column, colType);
        	}
        		
        }
    }

    /**
     * @param conn Connection
     * @param tableName Table name for insert statement.
     * @param allColumns All column names for this table.
     * @return PreparedStatement for Insert SQL commands.
     */
    private String generateInsertPreparedStatementSql(final String tableName, final List<String> allColumns) {

        final StringBuilder sqlStr = new StringBuilder();
        sqlStr.append("INSERT INTO ");
        sqlStr.append(tableName);
        sqlStr.append("(");
        boolean bFirst = true;
        for (String s : allColumns) {
            if (!bFirst) {
                sqlStr.append(",");
            }
            bFirst = false;
            sqlStr.append(s);
        }

        sqlStr.append(") VALUES (");
        bFirst = true;
        for (int i = 0; i < allColumns.size(); i++) {
            if (!bFirst) {
                sqlStr.append(",");
            }
            bFirst = false;
            sqlStr.append("?");
        }
        sqlStr.append(")");

        return sqlStr.toString();
    }

    private String generateDeletePreparedStatementSql(final String tableName, final List<String> primaryKeys) {
        final StringBuilder sqlStr = new StringBuilder();
        sqlStr.append("DELETE FROM ");
        sqlStr.append(tableName);
        sqlStr.append(" WHERE ");
        boolean bFirst = true;
        for (String s : primaryKeys) {
            if (!bFirst) {
                sqlStr.append(" AND ");
            }
            bFirst = false;
            sqlStr.append(s);
            sqlStr.append("=?");
        }
        return sqlStr.toString();
    }

    private String generateUpdatePreparedStatementSql(final String tableName, final List<String> primaryKeys, RowCompareResult result) {
        final StringBuilder sqlStr = new StringBuilder();
        sqlStr.append("UPDATE ");
        sqlStr.append(tableName);
        sqlStr.append(" SET ");
        boolean bFirst = true;
        for (String s : result.getChangedDataColumns()) {
            if (!bFirst) {
                sqlStr.append(",");
            }
            bFirst = false;
            sqlStr.append(s);
            sqlStr.append("=?");
        }
        sqlStr.append(" where ");
        bFirst = true;
        for (String s : primaryKeys) {
            if (!bFirst) {
                sqlStr.append(" AND ");
            }
            bFirst = false;
            sqlStr.append(s);
            sqlStr.append("=?");
        }
        return sqlStr.toString();    }


    /**
     * @param sourceMd
     * @param destMd
     * @param sourceRs
     * @param destRs
     * @param result
     * @param tableName
     * @param primaryKeys
     * @return
     * @throws SQLException
     */
/*    
    private String generateUpdateSql(DatabaseMetaData sourceMd, DatabaseMetaData destMd, ResultSet sourceRs, ResultSet destRs, RowCompareResult result, String tableName, List<String> primaryKeys) throws SQLException {
        final StringBuilder sqlStr = new StringBuilder();

        if (result.getResult() == RowCompareResult.UPDATE_DESTINATION_DATA) {
            sqlStr.append("update ");
            sqlStr.append(tableName);
            sqlStr.append(" set ");
            boolean bFirst = true;
            for (String s : result.getChangedDataColumns()) {
                if (!bFirst) {
                    sqlStr.append(",");
                }
                bFirst = false;
                sqlStr.append(s);
                sqlStr.append(translateColumnData(sourceRs.getObject(s), destMd.getDatabaseProductName(), true));
            }
            sqlStr.append(" where ");
            bFirst = true;
            for (String s : primaryKeys) {
                if (!bFirst) {
                    sqlStr.append(" AND ");
                }
                bFirst = false;
                sqlStr.append(s);
                sqlStr.append(translateColumnData(destRs.getObject(s), sourceMd.getDatabaseProductName(), true));
            }
        }
        return sqlStr.toString();
    }

    private static String translateColumnData(Object inputObj) {
        return translateColumnData(inputObj, null, false);
    }
*/
    /**
     * @param inputObj
     * @param sqlServerProductName
     * @param includeEqualSign
     * @return translated data
     */
/*    
    private static String translateColumnData(Object inputObj, String sqlServerProductName, boolean includeEqualSign) {
        if (inputObj == null && includeEqualSign) {
           return " IS NULL";
        }

        final StringBuilder retStr = new StringBuilder();
        try {
            boolean quotes = true;
            if (includeEqualSign) {
                retStr.append("=");
            }
            if (inputObj instanceof byte[]) {
                if (sqlServerProductName == null) {
                    quotes = true;
                } else if (sqlServerProductName.toUpperCase().contains("DERBY")) {
                    retStr.append("X");
                } else if (sqlServerProductName.toUpperCase().contains("MICROSOFT SQL SERVER")) {
                    retStr.append("0x");
                    quotes = false;
                } else if (sqlServerProductName.toUpperCase().contains("MYSQL")) {
                    retStr.append("x");
                } else if (sqlServerProductName.toUpperCase().contains("ORACLE")) {
                    quotes = true;
                }

                if (quotes) {
                    retStr.append("'");
                }
                retStr.append(DatabaseUtils.getHex((byte[]) inputObj));
                if (quotes) {
                    retStr.append("'");
                }
            } else if (inputObj instanceof java.sql.Timestamp) {
                retStr.append("{ts '");
                retStr.append(inputObj);
                retStr.append("'}");
            } else if (inputObj instanceof oracle.sql.TIMESTAMP) {
                retStr.append("{ts '");
                oracle.sql.TIMESTAMP ts = (oracle.sql.TIMESTAMP) inputObj;
                java.sql.Timestamp jts = ts.timestampValue();
                retStr.append(jts);
                retStr.append("'}");
            } else if (inputObj instanceof java.util.Date) {
                retStr.append("{ts '");
                retStr.append(inputObj);
                retStr.append("'}");
            } else if (inputObj instanceof String) {
                String str = ((String) inputObj).replace("'", "''");  //need to replace single ' with double '' so that they can be inserted properly.
                retStr.append("'");
                retStr.append(str);
                retStr.append("'");
            } else {
                retStr.append(inputObj);
            }
        }
        catch (SQLException e) {
            RAFGLogger.error(e, "Error translating data for " + sqlServerProductName + " " + inputObj.toString(), null);
        } finally {

        }
        return retStr.toString();
    }
*/
    /**
     * @param o1
     * @param o2
     * @param primaryKeys
     * @return RowCompareResult
     * @throws SQLException
     */
    private static RowCompareResult compareResultSets(ResultSet o1, ResultSet o2, List<String> primaryKeys) throws SQLException {
        boolean bDataColumnsMatch = true;

        ResultSetMetaData o1md = o1.getMetaData();
        ResultSetMetaData o2md = o2.getMetaData();

        int o1Size = o1md.getColumnCount();
        int o2Size = o2md.getColumnCount();

        if (o1Size != o2Size) {
            throw new ClassCastException("ResultSet1 and ResultSet2 returned a different number of columns from getColumnCount()");
        }

        RowCompareResult result = new RowCompareResult();

        for (int i = 1; i <= o1Size; i++) {
/*        	
            if (LOCAL_DBG) {
                RAFGLogger.info("Working on column:"
                        + " Left={[" + o1md.getColumnName(i) + "]; type: [" + o1md.getColumnType(i) + ":" + o1md.getColumnTypeName(i) + "]; value: [" + o1.getObject(i) + "]}"
                        + " Right={[" + o2md.getColumnName(i) + "]; type: [" + o2md.getColumnType(i) + ":" + o2md.getColumnTypeName(i) + "]; value: [" + o2.getObject(i) + "]}");
            }
*/            

            String colName = o1md.getColumnName(i).toUpperCase();

            int colCompResult = compareColumn(i, o1, o2, o1md, o2md);

            if (colCompResult < 0) { //source column < dest column
                if (primaryKeys.contains(colName)) {  //this column is a part of the primary key
                    result.setResult(RowCompareResult.INSERT_SOURCE_ROW);  //this row does not exist in destination, need to add it.
                    break;
                } else {
                    bDataColumnsMatch = false;  //this is a regular data column and they don't match.  need to flag that.
                    result.getChangedDataColumns().add(colName);
                }
            } else if (colCompResult > 0) { //source column > dest column
                if (primaryKeys.contains(colName)) {  //this column is a part of the primary key
                    result.setResult(RowCompareResult.DELETE_DESTINATION_ROW);  //this row does not exist in source, need to delete from destination.
                    break;
                } else {
                    bDataColumnsMatch = false;  //this is a regular data column and they don't match.  need to flag that.
                    result.getChangedDataColumns().add(colName);
                }
            }
        }
        if (!bDataColumnsMatch) {
            result.setResult(RowCompareResult.UPDATE_DESTINATION_DATA);
        }

        return result;
    }

    /**
     * @param colNbr
     * @param o1
     * @param o2
     * @param o1md
     * @param o2md
     * @return
     * @throws SQLException
     */
    private static int compareColumn(int colNbr, ResultSet o1, ResultSet o2, ResultSetMetaData o1md, ResultSetMetaData o2md) throws SQLException
    {
        //0 = equal
        //-1 = source col < dest col
        //1 = source col > dest col

        int i = colNbr;

        if (!o1md.getColumnName(i).equalsIgnoreCase(o2md.getColumnName(i))) {
            throw new ClassCastException("ResultSet1 and ResultSet2 have different column names for position " + i + ": " + o1md.getColumnName(i) + " vs " + o2md.getColumnName(i));
        }
    	int col1Type = o1md.getColumnType(i);
    	int col2Type = o2md.getColumnType(i);
//    	String col1TypeName = o1md.getColumnTypeName(i);
//    	String col2TypeName = o2md.getColumnTypeName(i);
        if (col1Type != col2Type) {
            // if source is a DATE format, and dest is TIMESTAMP, allow through.  Oracle seems to represent DATE as a timestamp.
            // if source is a TIMESTAMP format, and dest is DATE, allow through too.
        	// if source is DECIMAL, and dest is NUMERIC, thats okay too.
            if (!(col1Type ==  java.sql.Types.DATE && col2Type == java.sql.Types.TIMESTAMP)
            		&& !(col1Type ==  java.sql.Types.TIMESTAMP && col2Type == java.sql.Types.DATE)
            		&& !(col1Type == java.sql.Types.NUMERIC || col1Type == java.sql.Types.DECIMAL
            				&& col2Type == java.sql.Types.NUMERIC || col2Type == java.sql.Types.DECIMAL)) {
            	String tableName = o2md.getTableName(i);
                throw new ClassCastException(
                        "ResultSet1 and ResultSet2 column " + tableName + "." + o1md.getColumnName(i)
                        + " have different data types: source=" + DatabaseUtils.getJdbcTypeName(o1md.getColumnType(i))
                        + " vs dest=" + DatabaseUtils.getJdbcTypeName(o2md.getColumnType(i))
                        );
            }
        }

        int colType = o2md.getColumnType(i);


        int comp;
        byte[] b1 = null;
        byte[] b2 = null;
        switch(colType) {
            case java.sql.Types.VARCHAR:
            case java.sql.Types.CHAR:
            case java.sql.Types.LONGVARCHAR:
                String o1Str = o1.getString(i);
                String o2Str = o2.getString(i);
                if (o1Str != null && o2Str != null) {
                    comp = o1.getString(i).compareTo(o2.getString(i));
                } else if (o1Str == null && o2Str != null) {
                    comp = -1;
                } else if (o1Str != null && o2Str == null) {
                    comp = 1;
                } else {
                    comp = 0;
                }
                if (comp != 0) {
                    return comp;
                }
                break;
            case java.sql.Types.INTEGER:
                if (o1.getInt(i) < o2.getInt(i)) {
                    return -1;
                } else if (o1.getInt(i) > o2.getInt(i)) {
                    return 1;
                }
                break;
            case java.sql.Types.BIGINT:
                if (o1.getLong(i) < o2.getLong(i)) {
                    return -1;
                } else if (o1.getLong(i) > o2.getLong(i)) {
                    return 1;
                }
                break;
            case java.sql.Types.REAL:
            case java.sql.Types.FLOAT:
                if (o1.getFloat(i) < o2.getFloat(i)) {
                    return -1;
                } else if (o1.getFloat(i) > o2.getFloat(i)) {
                    return 1;
                }
                break;
            case java.sql.Types.DOUBLE:
                if (o1.getDouble(i) < o2.getDouble(i)) {
                    return -1;
                } else if (o1.getDouble(i) > o2.getDouble(i)) {
                    return 1;
                }
                break;
            case java.sql.Types.DECIMAL:
            case java.sql.Types.NUMERIC:
                BigDecimal bd1 = o1.getBigDecimal(i);
                BigDecimal bd2 = o2.getBigDecimal(i);
                if (bd1 == null && bd2 != null) {
                    return -1;
                } else if (bd1 != null && bd2 == null) {
                    return 1;
                } else if (bd1 == null && bd2 == null) {
                    return 0;
                } else {
                    return bd1.compareTo(bd2);
                }
            case java.sql.Types.BINARY:
                if (o1.getBytes(i) != null) {
                    b1 = o1.getBytes(i);
                }
                if (o2.getBytes(i) != null) {
                    b2 = o2.getBytes(i);
                }
                if (!Arrays.equals(b1, b2)) {
                    return -1;
                }
                break;
            case java.sql.Types.BIT:
                if (!o1.getBoolean(i) && o2.getBoolean(i)) {
                    return -1;
                } else if (o1.getBoolean(i) && !o2.getBoolean(i)) {
                    return 1;
                }
                break;
            case java.sql.Types.TINYINT:
            case java.sql.Types.SMALLINT:
                if (o1.getShort(i) < o2.getShort(i)) {
                    return -1;
                } else if (o1.getShort(i) > o2.getShort(i)) {
                    return 1;
                }
                break;
            case java.sql.Types.TIMESTAMP:
                java.sql.Timestamp o1ts = o1.getTimestamp(i);
                java.sql.Timestamp o2ts = o2.getTimestamp(i);
                if (o1ts != null && o2ts != null) {
                    comp = o1ts.compareTo(o2ts);
                } else if (o1ts == null && o2ts != null) {
                    comp = -1;
                } else if (o1ts != null && o2ts == null) {
                    comp = 1;
                } else {
                    comp = 0;
                }
                if (comp != 0) {
                    return comp;
                }
                break;
            case java.sql.Types.TIME:
                java.sql.Time o1t = o1.getTime(i);
                java.sql.Time o2t = o2.getTime(i);
                if (o1t != null && o2t != null) {
                    comp = o1t.compareTo(o2t);
                } else if (o1t == null && o2t != null) {
                    comp = -1;
                } else if (o1t != null && o2t == null) {
                    comp = 1;
                } else {
                    comp = 0;
                }
                if (comp != 0) {
                    return comp;
                }
                break;
            case java.sql.Types.DATE:
                java.sql.Date o1d = o1.getDate(i);
                java.sql.Date o2d = o2.getDate(i);
                if (o1d != null && o2d != null) {
                    comp = o1d.compareTo(o2d);
                } else if (o1d == null && o2d != null) {
                    comp = -1;
                } else if (o1d != null && o2d == null) {
                    comp = 1;
                } else {
                    comp = 0;
                }
                if (comp != 0) {
                    return comp;
                }
                break;
            case java.sql.Types.BLOB:
                java.sql.Blob o1blob = o1.getBlob(i);
                java.sql.Blob o2blob = o2.getBlob(i);
                if (o1blob != null && o1blob.length() > 0) {
                    b1 = o1blob.getBytes((long) 1, (int) o1blob.length());
                }
                if (o2blob != null && o2blob.length() > 0) {
                    b2 = o2blob.getBytes((long) 1, (int) o2blob.length());
                }
                if (!Arrays.equals(b1, b2)) {
                    return -1;
                }
                break;
            case java.sql.Types.CLOB:
                String c1 = null;
                java.sql.Clob o1clob = o1.getClob(i);
                if (o1clob != null && o1clob.length() > 0) {
                    c1 = o1clob.getSubString((long) 1, (int) o1clob.length());
                }
                String c2 = null;
                java.sql.Clob o2clob = o2.getClob(i);
                if (o2clob != null && o2clob.length() > 0) {
                    c2 = o2clob.getSubString((long) 1, (int) o2clob.length());
                }
                if (!c1.equals(c2)) {
                    return -1;
                }
                break;
            default:
                throw new ClassCastException("Don't know how to handle SQL data type " + o1md.getColumnTypeName(i) + " for column " + o1md.getColumnName(i));
        }

        return 0;
    }

    /**
     * @author RAFG Development Team (D9127520)
     *
     */
    private static class RowCompareResult {
        /**
         *
         */
        public static final short UPDATE_DESTINATION_DATA = 0;
        /**
         *
         */
        public static final short INSERT_SOURCE_ROW = 1;
        /**
         *
         */
        public static final short DELETE_DESTINATION_ROW = 2;
        /**
         *
         */
        public static final short ROWS_MATCH = 3;

        /**
         *
         */
        private short result = ROWS_MATCH;
        /**
         *
         */
        private List<String> changedDataColumns = null;

        /**
         *
         */
        public RowCompareResult() {
            super();
        }

        /**
         * @param result
         */
        public RowCompareResult(short result) {
            this.result = result;
        }

        /**
         * @return changed data columns
         */
        public synchronized List<String> getChangedDataColumns() {
            if (changedDataColumns == null) {
                changedDataColumns = new LinkedList<String>();
            }
            return changedDataColumns;
        }
        /**
         * @return result
         */
        public short getResult() {
            return result;
        }

        /**
         * @param result the result
         */
        public void setResult(short result) {
            this.result = result;
        }
    }

}
